//
// Copyright 2010 The Android Open Source Project
//
// A looper implementation based on epoll().
//
#define LOG_TAG "Looper"

//#define LOG_NDEBUG 0

// Debugs poll and wake interactions.
#define DEBUG_POLL_AND_WAKE 0

// Debugs callback registration and invocation.
#define DEBUG_CALLBACKS 0

#include <errno.h>
#include <fcntl.h>
#include <sys/eventfd.h>

#include <utils/Looper.h>

namespace android {

// --- WeakMessageHandler ---

    WeakMessageHandler::WeakMessageHandler(const wp<MessageHandler> &handler) :
            mHandler(handler) {
    }

    WeakMessageHandler::~WeakMessageHandler() {
    }

    void WeakMessageHandler::handleMessage(const Message &message) {
        sp<MessageHandler> handler = mHandler.promote();
        if (handler != NULL) {
            handler->handleMessage(message);
        }
    }


// --- SimpleLooperCallback ---

    SimpleLooperCallback::SimpleLooperCallback(Looper_callbackFunc callback) :
            mCallback(callback) {
    }

    SimpleLooperCallback::~SimpleLooperCallback() {
    }

    int SimpleLooperCallback::handleEvent(int fd, int events, void *data) {
        return mCallback(fd, events, data);
    }


// --- Looper ---

// Hint for number of file descriptors to be associated with the epoll instance.
    static const int EPOLL_SIZE_HINT = 8;

// Maximum number of file descriptors for which to retrieve poll events each iteration.
    static const int EPOLL_MAX_EVENTS = 16;

    static pthread_once_t gTLSOnce = PTHREAD_ONCE_INIT;
    static pthread_key_t gTLSKey = 0;

    Looper::Looper(bool allowNonCallbacks) :
            mAllowNonCallbacks(allowNonCallbacks), mSendingMessage(false),
            mPolling(false), mEpollFd(-1), mEpollRebuildRequired(false),
            mNextRequestSeq(0), mResponseIndex(0), mNextMessageUptime(LLONG_MAX) {
        mWakeEventFd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
        LOG_ALWAYS_FATAL_IF(mWakeEventFd < 0, "Could not make wake event fd: %s",
                            strerror(errno));

        AutoMutex _l(mLock);
        rebuildEpollLocked();
    }

    Looper::~Looper() {
        close(mWakeEventFd);
        if (mEpollFd >= 0) {
            close(mEpollFd);
        }
    }

    void Looper::initTLSKey() {
        int result = pthread_key_create(&gTLSKey, threadDestructor);
        LOG_ALWAYS_FATAL_IF(result != 0, "Could not allocate TLS key.");
    }

    void Looper::threadDestructor(void *st) {
        Looper *const self = static_cast<Looper *>(st);
        if (self != NULL) {
            self->decStrong((void *) threadDestructor);
        }
    }

    void Looper::setForThread(const sp<Looper> &looper) {
        sp<Looper> old = getForThread(); // also has side-effect of initializing TLS

        if (looper != NULL) {
            looper->incStrong((void *) threadDestructor);
        }

        pthread_setspecific(gTLSKey, looper.get());

        if (old != NULL) {
            old->decStrong((void *) threadDestructor);
        }
    }

    sp<Looper> Looper::getForThread() {
        int result = pthread_once(&gTLSOnce, initTLSKey);
        LOG_ALWAYS_FATAL_IF(result != 0, "pthread_once failed");

        return (Looper *) pthread_getspecific(gTLSKey);
    }

    sp<Looper> Looper::prepare(int opts) {
        bool allowNonCallbacks = opts & PREPARE_ALLOW_NON_CALLBACKS;
        sp<Looper> looper = Looper::getForThread();
        if (looper == NULL) {
            looper = new Looper(allowNonCallbacks);
            Looper::setForThread(looper);
        }
        if (looper->getAllowNonCallbacks() != allowNonCallbacks) {
            ALOGW("Looper already prepared for this thread with a different value for the "
                          "LOOPER_PREPARE_ALLOW_NON_CALLBACKS option.");
        }
        return looper;
    }

    bool Looper::getAllowNonCallbacks() const {
        return mAllowNonCallbacks;
    }

    void Looper::rebuildEpollLocked() {
        // Close old epoll instance if we have one.
        if (mEpollFd >= 0) {
#if DEBUG_CALLBACKS
            ALOGD("%p ~ rebuildEpollLocked - rebuilding epoll set", this);
#endif
            close(mEpollFd);
        }

        // Allocate the new epoll instance and register the wake pipe.
        mEpollFd = epoll_create(EPOLL_SIZE_HINT);
        LOG_ALWAYS_FATAL_IF(mEpollFd < 0, "Could not create epoll instance: %s", strerror(errno));

        struct epoll_event eventItem;
        memset(&eventItem, 0, sizeof(epoll_event)); // zero out unused members of data field union
        eventItem.events = EPOLLIN;
        eventItem.data.fd = mWakeEventFd;
        int result = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, mWakeEventFd, &eventItem);
        LOG_ALWAYS_FATAL_IF(result != 0, "Could not add wake event fd to epoll instance: %s",
                            strerror(errno));

        for (size_t i = 0; i < mRequests.size(); i++) {
            const Request &request = mRequests.valueAt(i);
            struct epoll_event eventItem;
            request.initEventItem(&eventItem);

            int epollResult = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, request.fd, &eventItem);
            if (epollResult < 0) {
                LOGE("Error adding epoll events for fd %d while rebuilding epoll set: %s",
                     request.fd, strerror(errno));
            }
        }
    }

    void Looper::scheduleEpollRebuildLocked() {
        if (!mEpollRebuildRequired) {
#if DEBUG_CALLBACKS
            ALOGD("%p ~ scheduleEpollRebuildLocked - scheduling epoll set rebuild", this);
#endif
            mEpollRebuildRequired = true;
            wake();
        }
    }

    int Looper::pollOnce(int timeoutMillis, int *outFd, int *outEvents, void **outData) {
        int result = 0;
        for (;;) {
            while (mResponseIndex < mResponses.size()) {
                const Response &response = mResponses.itemAt(mResponseIndex++);
                int ident = response.request.ident;
                if (ident >= 0) {
                    int fd = response.request.fd;
                    int events = response.events;
                    void *data = response.request.data;
#if DEBUG_POLL_AND_WAKE
                    ALOGD("%p ~ pollOnce - returning signalled identifier %d: "
                            "fd=%d, events=0x%x, data=%p",
                            this, ident, fd, events, data);
#endif
                    if (outFd != NULL) *outFd = fd;
                    if (outEvents != NULL) *outEvents = events;
                    if (outData != NULL) *outData = data;
                    return ident;
                }
            }

            if (result != 0) {
#if DEBUG_POLL_AND_WAKE
                ALOGD("%p ~ pollOnce - returning result %d", this, result);
#endif
                if (outFd != NULL) *outFd = 0;
                if (outEvents != NULL) *outEvents = 0;
                if (outData != NULL) *outData = NULL;
                return result;
            }

            result = pollInner(timeoutMillis);
        }
    }

    int Looper::pollInner(int timeoutMillis) {
#if DEBUG_POLL_AND_WAKE
        ALOGD("%p ~ pollOnce - waiting: timeoutMillis=%d", this, timeoutMillis);
#endif

        // Adjust the timeout based on when the next message is due.
        if (timeoutMillis != 0 && mNextMessageUptime != LLONG_MAX) {
            nsecs_t now = systemTime(SYSTEM_TIME_MONOTONIC);
            int messageTimeoutMillis = toMillisecondTimeoutDelay(now, mNextMessageUptime);
            if (messageTimeoutMillis >= 0
                && (timeoutMillis < 0 || messageTimeoutMillis < timeoutMillis)) {
                timeoutMillis = messageTimeoutMillis;
            }
#if DEBUG_POLL_AND_WAKE
            ALOGD("%p ~ pollOnce - next message in %" PRId64 "ns, adjusted timeout: timeoutMillis=%d",
                    this, mNextMessageUptime - now, timeoutMillis);
#endif
        }

        // Poll.
        int result = POLL_WAKE;
        mResponses.clear();
        mResponseIndex = 0;

        // We are about to idle.
        mPolling = true;

        struct epoll_event eventItems[EPOLL_MAX_EVENTS];
        int eventCount = epoll_wait(mEpollFd, eventItems, EPOLL_MAX_EVENTS, timeoutMillis);

        // No longer idling.
        mPolling = false;

        // Acquire lock.
        mLock.lock();

        // Rebuild epoll set if needed.
        if (mEpollRebuildRequired) {
            mEpollRebuildRequired = false;
            rebuildEpollLocked();
            goto Done;
        }

        // Check for poll error.
        if (eventCount < 0) {
            if (errno == EINTR) {
                goto Done;
            }
            ALOGW("Poll failed with an unexpected error: %s", strerror(errno));
            result = POLL_ERROR;
            goto Done;
        }

        // Check for poll timeout.
        if (eventCount == 0) {
#if DEBUG_POLL_AND_WAKE
            ALOGD("%p ~ pollOnce - timeout", this);
#endif
            result = POLL_TIMEOUT;
            goto Done;
        }

        // Handle all events.
#if DEBUG_POLL_AND_WAKE
        ALOGD("%p ~ pollOnce - handling events from %d fds", this, eventCount);
#endif

        for (int i = 0; i < eventCount; i++) {
            int fd = eventItems[i].data.fd;
            uint32_t epollEvents = eventItems[i].events;
            if (fd == mWakeEventFd) {
                if (epollEvents & EPOLLIN) {
                    awoken();
                } else {
                    ALOGW("Ignoring unexpected epoll events 0x%x on wake event fd.", epollEvents);
                }
            } else {
                ssize_t requestIndex = mRequests.indexOfKey(fd);
                if (requestIndex >= 0) {
                    int events = 0;
                    if (epollEvents & EPOLLIN) events |= EVENT_INPUT;
                    if (epollEvents & EPOLLOUT) events |= EVENT_OUTPUT;
                    if (epollEvents & EPOLLERR) events |= EVENT_ERROR;
                    if (epollEvents & EPOLLHUP) events |= EVENT_HANGUP;
                    pushResponse(events, mRequests.valueAt(requestIndex));
                } else {
                    ALOGW("Ignoring unexpected epoll events 0x%x on fd %d that is "
                                  "no longer registered.", epollEvents, fd);
                }
            }
        }
        Done:;

        // Invoke pending message callbacks.
        mNextMessageUptime = LLONG_MAX;
        while (mMessageEnvelopes.size() != 0) {
            nsecs_t now = systemTime(SYSTEM_TIME_MONOTONIC);
            const MessageEnvelope &messageEnvelope = mMessageEnvelopes.itemAt(0);
            if (messageEnvelope.uptime <= now) {
                // Remove the envelope from the list.
                // We keep a strong reference to the handler until the call to handleMessage
                // finishes.  Then we drop it so that the handler can be deleted *before*
                // we reacquire our lock.
                { // obtain handler
                    sp<MessageHandler> handler = messageEnvelope.handler;
                    Message message = messageEnvelope.message;
                    mMessageEnvelopes.removeAt(0);
                    mSendingMessage = true;
                    mLock.unlock();

#if DEBUG_POLL_AND_WAKE || DEBUG_CALLBACKS
                    ALOGD("%p ~ pollOnce - sending message: handler=%p, what=%d",
                            this, handler.get(), message.what);
#endif
                    handler->handleMessage(message);
                } // release handler

                mLock.lock();
                mSendingMessage = false;
                result = POLL_CALLBACK;
            } else {
                // The last message left at the head of the queue determines the next wakeup time.
                mNextMessageUptime = messageEnvelope.uptime;
                break;
            }
        }

        // Release lock.
        mLock.unlock();

        // Invoke all response callbacks.
        for (size_t i = 0; i < mResponses.size(); i++) {
            Response &response = mResponses.editItemAt(i);
            if (response.request.ident == POLL_CALLBACK) {
                int fd = response.request.fd;
                int events = response.events;
                void *data = response.request.data;
#if DEBUG_POLL_AND_WAKE || DEBUG_CALLBACKS
                ALOGD("%p ~ pollOnce - invoking fd event callback %p: fd=%d, events=0x%x, data=%p",
                        this, response.request.callback.get(), fd, events, data);
#endif
                // Invoke the callback.  Note that the file descriptor may be closed by
                // the callback (and potentially even reused) before the function returns so
                // we need to be a little careful when removing the file descriptor afterwards.
                int callbackResult = response.request.callback->handleEvent(fd, events, data);
                if (callbackResult == 0) {
                    removeFd(fd, response.request.seq);
                }

                // Clear the callback reference in the response structure promptly because we
                // will not clear the response vector itself until the next poll.
                response.request.callback.clear();
                result = POLL_CALLBACK;
            }
        }
        return result;
    }

    int Looper::pollAll(int timeoutMillis, int *outFd, int *outEvents, void **outData) {
        if (timeoutMillis <= 0) {
            int result;
            do {
                result = pollOnce(timeoutMillis, outFd, outEvents, outData);
            } while (result == POLL_CALLBACK);
            return result;
        } else {
            nsecs_t endTime = systemTime(SYSTEM_TIME_MONOTONIC)
                              + milliseconds_to_nanoseconds(timeoutMillis);

            for (;;) {
                int result = pollOnce(timeoutMillis, outFd, outEvents, outData);
                if (result != POLL_CALLBACK) {
                    return result;
                }

                nsecs_t now = systemTime(SYSTEM_TIME_MONOTONIC);
                timeoutMillis = toMillisecondTimeoutDelay(now, endTime);
                if (timeoutMillis == 0) {
                    return POLL_TIMEOUT;
                }
            }
        }
    }

    void Looper::wake() {
#if DEBUG_POLL_AND_WAKE
        ALOGD("%p ~ wake", this);
#endif

        uint64_t inc = 1;
        ssize_t nWrite = TEMP_FAILURE_RETRY(write(mWakeEventFd, &inc, sizeof(uint64_t)));
        if (nWrite != sizeof(uint64_t)) {
            if (errno != EAGAIN) {
                ALOGW("Could not write wake signal: %s", strerror(errno));
            }
        }
    }

    void Looper::awoken() {
#if DEBUG_POLL_AND_WAKE
        ALOGD("%p ~ awoken", this);
#endif

        uint64_t counter;
        TEMP_FAILURE_RETRY(read(mWakeEventFd, &counter, sizeof(uint64_t)));
    }

    void Looper::pushResponse(int events, const Request &request) {
        Response response;
        response.events = events;
        response.request = request;
        mResponses.push(response);
    }

    int Looper::addFd(int fd, int ident, int events, Looper_callbackFunc callback, void *data) {
        return addFd(fd, ident, events, callback ? new SimpleLooperCallback(callback) : NULL, data);
    }

    int
    Looper::addFd(int fd, int ident, int events, const sp<LooperCallback> &callback, void *data) {
#if DEBUG_CALLBACKS
        ALOGD("%p ~ addFd - fd=%d, ident=%d, events=0x%x, callback=%p, data=%p", this, fd, ident,
                events, callback.get(), data);
#endif

        if (!callback.get()) {
            if (!mAllowNonCallbacks) {
                LOGE("Invalid attempt to set NULL callback but not allowed for this looper.");
                return -1;
            }

            if (ident < 0) {
                LOGE("Invalid attempt to set NULL callback with ident < 0.");
                return -1;
            }
        } else {
            ident = POLL_CALLBACK;
        }

        { // acquire lock
            AutoMutex _l(mLock);

            Request request;
            request.fd = fd;
            request.ident = ident;
            request.events = events;
            request.seq = mNextRequestSeq++;
            request.callback = callback;
            request.data = data;
            if (mNextRequestSeq == -1) mNextRequestSeq = 0; // reserve sequence number -1

            struct epoll_event eventItem;
            request.initEventItem(&eventItem);

            ssize_t requestIndex = mRequests.indexOfKey(fd);
            if (requestIndex < 0) {
                int epollResult = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, fd, &eventItem);
                if (epollResult < 0) {
                    LOGE("Error adding epoll events for fd %d: %s", fd, strerror(errno));
                    return -1;
                }
                mRequests.add(fd, request);
            } else {
                int epollResult = epoll_ctl(mEpollFd, EPOLL_CTL_MOD, fd, &eventItem);
                if (epollResult < 0) {
                    if (errno == ENOENT) {
                        // Tolerate ENOENT because it means that an older file descriptor was
                        // closed before its callback was unregistered and meanwhile a new
                        // file descriptor with the same number has been created and is now
                        // being registered for the first time.  This error may occur naturally
                        // when a callback has the side-effect of closing the file descriptor
                        // before returning and unregistering itself.  Callback sequence number
                        // checks further ensure that the race is benign.
                        //
                        // Unfortunately due to kernel limitations we need to rebuild the epoll
                        // set from scratch because it may contain an old file handle that we are
                        // now unable to remove since its file descriptor is no longer valid.
                        // No such problem would have occurred if we were using the poll system
                        // call instead, but that approach carries others disadvantages.
#if DEBUG_CALLBACKS
                        ALOGD("%p ~ addFd - EPOLL_CTL_MOD failed due to file descriptor "
                                "being recycled, falling back on EPOLL_CTL_ADD: %s",
                                this, strerror(errno));
#endif
                        epollResult = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, fd, &eventItem);
                        if (epollResult < 0) {
                            LOGE("Error modifying or adding epoll events for fd %d: %s",
                                 fd, strerror(errno));
                            return -1;
                        }
                        scheduleEpollRebuildLocked();
                    } else {
                        LOGE("Error modifying epoll events for fd %d: %s", fd, strerror(errno));
                        return -1;
                    }
                }
                mRequests.replaceValueAt(requestIndex, request);
            }
        } // release lock
        return 1;
    }

    int Looper::removeFd(int fd) {
        return removeFd(fd, -1);
    }

    int Looper::removeFd(int fd, int seq) {
#if DEBUG_CALLBACKS
        ALOGD("%p ~ removeFd - fd=%d, seq=%d", this, fd, seq);
#endif

        { // acquire lock
            AutoMutex _l(mLock);
            ssize_t requestIndex = mRequests.indexOfKey(fd);
            if (requestIndex < 0) {
                return 0;
            }

            // Check the sequence number if one was given.
            if (seq != -1 && mRequests.valueAt(requestIndex).seq != seq) {
#if DEBUG_CALLBACKS
                ALOGD("%p ~ removeFd - sequence number mismatch, oldSeq=%d",
                        this, mRequests.valueAt(requestIndex).seq);
#endif
                return 0;
            }

            // Always remove the FD from the request map even if an error occurs while
            // updating the epoll set so that we avoid accidentally leaking callbacks.
            mRequests.removeItemsAt(requestIndex);

            int epollResult = epoll_ctl(mEpollFd, EPOLL_CTL_DEL, fd, NULL);
            if (epollResult < 0) {
                if (seq != -1 && (errno == EBADF || errno == ENOENT)) {
                    // Tolerate EBADF or ENOENT when the sequence number is known because it
                    // means that the file descriptor was closed before its callback was
                    // unregistered.  This error may occur naturally when a callback has the
                    // side-effect of closing the file descriptor before returning and
                    // unregistering itself.
                    //
                    // Unfortunately due to kernel limitations we need to rebuild the epoll
                    // set from scratch because it may contain an old file handle that we are
                    // now unable to remove since its file descriptor is no longer valid.
                    // No such problem would have occurred if we were using the poll system
                    // call instead, but that approach carries others disadvantages.
#if DEBUG_CALLBACKS
                    ALOGD("%p ~ removeFd - EPOLL_CTL_DEL failed due to file descriptor "
                            "being closed: %s", this, strerror(errno));
#endif
                    scheduleEpollRebuildLocked();
                } else {
                    // Some other error occurred.  This is really weird because it means
                    // our list of callbacks got out of sync with the epoll set somehow.
                    // We defensively rebuild the epoll set to avoid getting spurious
                    // notifications with nowhere to go.
                    LOGE("Error removing epoll events for fd %d: %s", fd, strerror(errno));
                    scheduleEpollRebuildLocked();
                    return -1;
                }
            }
        } // release lock
        return 1;
    }

    void Looper::sendMessage(const sp<MessageHandler> &handler, const Message &message) {
        nsecs_t now = systemTime(SYSTEM_TIME_MONOTONIC);
        sendMessageAtTime(now, handler, message);
    }

    void Looper::sendMessageDelayed(nsecs_t uptimeDelay, const sp<MessageHandler> &handler,
                                    const Message &message) {
        nsecs_t now = systemTime(SYSTEM_TIME_MONOTONIC);
        sendMessageAtTime(now + uptimeDelay, handler, message);
    }

    void Looper::sendMessageAtTime(nsecs_t uptime, const sp<MessageHandler> &handler,
                                   const Message &message) {
#if DEBUG_CALLBACKS
        ALOGD("%p ~ sendMessageAtTime - uptime=%" PRId64 ", handler=%p, what=%d",
                this, uptime, handler.get(), message.what);
#endif

        size_t i = 0;
        { // acquire lock
            AutoMutex _l(mLock);

            size_t messageCount = mMessageEnvelopes.size();
            while (i < messageCount && uptime >= mMessageEnvelopes.itemAt(i).uptime) {
                i += 1;
            }

            MessageEnvelope messageEnvelope(uptime, handler, message);
            mMessageEnvelopes.insertAt(messageEnvelope, i, 1);

            // Optimization: If the Looper is currently sending a message, then we can skip
            // the call to wake() because the next thing the Looper will do after processing
            // messages is to decide when the next wakeup time should be.  In fact, it does
            // not even matter whether this code is running on the Looper thread.
            if (mSendingMessage) {
                return;
            }
        } // release lock

        // Wake the poll loop only when we enqueue a new message at the head.
        if (i == 0) {
            wake();
        }
    }

    void Looper::removeMessages(const sp<MessageHandler> &handler) {
#if DEBUG_CALLBACKS
        ALOGD("%p ~ removeMessages - handler=%p", this, handler.get());
#endif

        { // acquire lock
            AutoMutex _l(mLock);

            for (size_t i = mMessageEnvelopes.size(); i != 0;) {
                const MessageEnvelope &messageEnvelope = mMessageEnvelopes.itemAt(--i);
                if (messageEnvelope.handler == handler) {
                    mMessageEnvelopes.removeAt(i);
                }
            }
        } // release lock
    }

    void Looper::removeMessages(const sp<MessageHandler> &handler, int what) {
#if DEBUG_CALLBACKS
        ALOGD("%p ~ removeMessages - handler=%p, what=%d", this, handler.get(), what);
#endif

        { // acquire lock
            AutoMutex _l(mLock);

            for (size_t i = mMessageEnvelopes.size(); i != 0;) {
                const MessageEnvelope &messageEnvelope = mMessageEnvelopes.itemAt(--i);
                if (messageEnvelope.handler == handler
                    && messageEnvelope.message.what == what) {
                    mMessageEnvelopes.removeAt(i);
                }
            }
        } // release lock
    }

    bool Looper::isPolling() const {
        return mPolling;
    }

    void Looper::Request::initEventItem(struct epoll_event *eventItem) const {
        int epollEvents = 0;
        if (events & EVENT_INPUT) epollEvents |= EPOLLIN;
        if (events & EVENT_OUTPUT) epollEvents |= EPOLLOUT;

        memset(eventItem, 0, sizeof(epoll_event)); // zero out unused members of data field union
        eventItem->events = epollEvents;
        eventItem->data.fd = fd;
    }

    MessageHandler::~MessageHandler() {}

    LooperCallback::~LooperCallback() {}

} // namespace android
